package com.lvmama.rhino.common.entity

import java.util.Date

import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import com.lvmama.rhino.analyze.client.PerformanceMonitor.logger
import com.lvmama.rhino.common.utils.{Constants, Utils}
import com.lvmama.rhino.common.utils.logging.LazyLogging
import net.minidev.json.JSONObject
import org.apache.spark.sql.Row
import spray.json.JsonParser

import scala.collection.JavaConversions.mapAsJavaMap
import scala.collection.immutable.HashMap.HashTrieMap
import scala.collection.mutable
import scala.util.parsing.json.JSON

/**
  * Created by yuanxiaofeng on 2016/1/28.
  */
case class BaseLog(logType: String, timeStamp: Long, firstChannel: String, secondChannel: String,
                   udid: String, osVersion: String, userAgent: String, userId: Long, ip: String,
                   lvversion: String, longitude: Float, latitude: Float, lvsessionid: String,
                   adChannel: String, pageCode: String, lastPageCode: String, productType: String)

/**
  * 接口服务器日志样本类
  *
  * @param logType
  * @param timeStamp       服务器收到request的时间
  * @param firstChannel
  * @param secondChannel
  * @param udid            设备唯一标示
  * @param osVersion       操作系统版本
  * @param userAgent       用户代理
  * @param userId
  * @param ip
  * @param lvversion       客户端版本
  * @param longitude
  * @param latitude
  * @param lvsessionid     客户端会话ID
  * @param adChannel       广告渠道
  * @param method
  * @param version         接口版本号
  * @param pageCode        当前页面code
  * @param lastPageCode    上步页面code
  * @param productType
  * @param clientTimestamp 客户端发起请求的时间戳(毫秒)
  */
case class ApiLog(logType: String, timeStamp: String, firstChannel: String, secondChannel: String,
                  udid: String, osVersion: String, userAgent: String, userId: String, ip: String,
                  lvversion: String, longitude: String, latitude: String, lvsessionid: String,
                  adChannel: String, method: String, version: String, pageCode: String, lastPageCode: String,
                  productType: String, clientTimestamp: String)


//trait NginxLog{
//  def apply(op: Option[Map[String, String]]): NginxLog
//  def unapply(arg: String): Option[Map[String, String]]
//}
/**
  * 花样男神nginx服务器日志样本类
  *
  * @param logType   日志类型
  * @param timestamp 时间戳
  * @param openId    微信授权id
  * @param pageCode  当前页面code
  * @param refer     上级页面code
  * @param eventId   事件Id
  */
case class NginxMan(logType: String, timestamp: String, openId: String, pageCode: String, refer: String, eventId: String)

object NginxMan {

  def apply(op: Option[Map[String, String]]): NginxMan = {
    val par = op.get
    new NginxMan(par.getOrElse("lt", ""), par.getOrElse("ts", ""), par.getOrElse("oi", ""), par.getOrElse("pc", ""), par.getOrElse("re", ""), par.getOrElse("ei", ""))
  }

  def unapply(arg: String): Option[Map[String, String]] = {
    val parts = arg.split("&")
    if (parts.nonEmpty) {
      val parMap = mutable.Map.empty[String, String]
      for (p <- parts) {
        val pair = p.split("=")
        try {
          parMap += (pair(0) -> pair(1))
        } catch {
          case e: ArrayIndexOutOfBoundsException => parMap += (pair(0) -> "")
        }
      }
      //      Some(parMap.getOrElse("lt", ""), parMap.getOrElse("ts", ""), parMap.getOrElse("oi", ""),parMap.getOrElse("pc", ""),parMap.getOrElse("re", ""),parMap.getOrElse("ei", ""))
      Some(parMap.toMap)
    }
    else None
  }
}

/**
  *
  * @param productType
  * @param productId
  * @param goodsId
  * @param udid
  * @param first_channel
  * @param off_on
  * @param lvsessionId
  */
case class MsyLog(productType: String, productId: String, goodsId: String, udid: String, first_channel: String, off_on: String, lvsessionId: String)

object MsyLog {
  def apply(op: Option[Map[String, String]]): MsyLog = {
    if (op.isDefined) {
      val par = op.get
      new MsyLog(par.getOrElse("productType", ""), par.getOrElse("productId", ""), par.getOrElse("goodsId", ""), par.getOrElse("udid", ""), par.getOrElse("first_channel", ""), par.getOrElse("off_on", ""), par.getOrElse("lvsessionId", ""))
    } else {
      new MsyLog("", "", "", "", "", "", "")
    }
  }

  def unapply(arg: String): Option[Map[String, String]] = {
    val parts = arg.split("&")
    if (parts.nonEmpty) {
      val parMap = mutable.Map.empty[String, String]
      for (p <- parts) {
        val pair = p.split("=")
        try {
          parMap += (pair(0) -> pair(1))
        } catch {
          case e: ArrayIndexOutOfBoundsException => parMap += (pair(0) -> "")
        }
      }
      //      Some(parMap.getOrElse("productType", ""), parMap.getOrElse("productId", ""), parMap.getOrElse("goodsId", ""), parMap.getOrElse("udid", ""))
      Some(parMap.toMap)
    } else
      None
  }

}


/**
  * @param lt 日志类型
  * @param ts 时间戳
  * @param ad 地址
  * @param la 经度
  * @param lo 纬度
  * @param av 客户端版本号
  * @param mm 客户端机型
  * @param dt 设备号
  * @param pc 当前页面
  * @param ei 事件id
  * @param qn 常见问题名
  */
case class AIMachineReceipt(lt: String, ts: String, ad: String, la: String,
                            lo: String, av: String, mm: String, dt: String,
                            pc: String, ei: String, qn: String)

object AIMachineReceipt {
  def apply(op: Option[(String, String, String, String, String, String, String,
    String, String, String, String)]): AIMachineReceipt = {
    if (op.isDefined) {
      val par = op.get
      new AIMachineReceipt(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9, par._10, par._11)
    } else {
      new AIMachineReceipt("", "", "", "", "", "", "", "", "", "", "")
    }
  }

  def unapply(arg: String): Option[(String, String, String, String, String, String, String,
    String, String, String, String)] = {
    val parts = arg.split(Constants.LOG_SEPERATOR_1)
    if (parts.nonEmpty && parts.length == 11) {
      Some(parts(0), parts(1), parts(2), parts(3), parts(4), parts(5), parts(6), parts(7),
        parts(8), parts(9), emojiFilter(parts(10)))
    } else
      None
  }

  //过滤掉emoji表情
  val emojiFilter: String => String = str => {
    if (!str.isEmpty && !" ".equals(str)) {
      val sb: StringBuilder = new StringBuilder
      for (i <- 0 until str.length) {
        val ch: Char = str.charAt(i)
        if (isEmojiCharacter(ch)) {
          sb.append(ch)
        }
      }
      sb.toString()
    } else {
      str
    }
  }

  lazy val isEmojiCharacter: Char => Boolean = ch =>
    ((ch == 0x0) || (ch == 0x9) || (ch == 0xA) || (ch == 0xD)
      || ((ch >= 0x20) && (ch <= 0xD7FF)) || ((ch >= 0xE000) && (ch <= 0xFFFD))
      || ((ch >= 0x10000) && (ch <= 0x10FFFF)))

}


/**
  *
  * @param tailCode         losc码
  * @param udid             设备号
  * @param platform         平台
  * @param messageGetTime   消息接收时间
  * @param messageClickTime 消息点击时间
  * @param getFlag          接收标记
  * @param clickFlag        点击标记
  * @param orderFlag        下单标记
  * @param payFlag          支付标记
  * @param orderId          订单id
  */
case class PushMessageReceipt(tailCode: String, udid: String, platform: String,
                              messageGetTime: String, messageClickTime: String, getFlag: String,
                              clickFlag: String, orderFlag: String, payFlag: String, orderId: String)

/**
  * 无线客户端监控日志
  *
  * @param logType      日志类型
  * @param deviceToken  设备号
  * @param operator     运营商
  * @param netType      网络类型
  * @param longitude    经度
  * @param latitude     纬度
  * @param province     省份
  * @param userSource   登录账号来源
  * @param userId       账号ID
  * @param buttonCode   按钮code
  * @param pageCode     页面code
  * @param channelCode  渠道code
  * @param platformCode 设备类型code
  * @param pageTypeCode 页面类型code
  * @param pageParam    具体页面参数
  *                     //  * @param pageAttribute 页面属性H5，非H5  统计二期删除
  * @param timestamp    时间戳
  *                     二期
  * @param stationId    站点
  * @param holeCode     坑位code
  * @param accessCode   访问入口code
  * @param ip           ip地址
  */
case class WirelessClientMonitorLog(logType: String, deviceToken: String, operator: String, netType: String, sessionId: String,
                                    longitude: String, latitude: String, province: String, userSource: String,
                                    userId: String, buttonCode: String, pageCode: String, channelCode: String, platformCode: String, pageTypeCode: String,
                                    pageParam: Map[String, String], timestamp: Long, stationId: String,
                                    holeCode: String, accessCode: String, newVisitor: String, ip: String)

object WirelessClientMonitorLog extends LazyLogging {
  def apply(op: Option[(String, String, String, String, String, String, String, String, String, String, String, String, String, String,
    String, Map[String, String], Long, String, String, String, String, String)]): WirelessClientMonitorLog = {
    if (op.isDefined) {
      val par = op.get
      new WirelessClientMonitorLog(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9, par._10,
        par._11, par._12, par._13, par._14, par._15, par._16, par._17, par._18, par._19, par._20, par._21, par._22)
    } else {
      new WirelessClientMonitorLog("", "", "", "", "", "", "", "", "", "", "", "", "", "", "", Map.empty[String, String],
        0L, "", "", "", "", "")

    }
  }

  def apply(): WirelessClientMonitorLog = {
    this.apply("", "", "", "", "", "", "", "", "", "", "", "", "", "", "", Map.empty[String, String], 0L, "", "", "", "", "")
  }

  def unapply(arg: String, ip: String, pageUrl: String): Option[(String, String, String, String, String, String, String, String, String, String, String, String, String,
    String, String, Map[String, String], Long, String, String, String, String, String)] = {
    try {
//      logger.info("parse begin:"+arg)
      val parseresult=JSON.parseFull(arg)
      parseresult match {
        case Some(parMap: HashTrieMap[String, String]) =>
          val pc: Array[String] = parMap.get("pc").get.split("\\.")
          val pageCode: Array[String] = if (parMap.get("pc").get.endsWith(".")) pc :+ "null" else pc
          val channelCode = pageCode.slice(0, 1).head
          val platformCode = pageCode.slice(1, 2).head.trim
          var entranceCode = "kwnull-null"
          var pageTypeCode = ""
          val topicPattern = "(3zhTi.*)".r
          val channelPattern = "(3PinD.*)".r
          var regex = "\\d+".r

          def subField(s: String) = {
            val endIndex =
              if (s.contains('?'))
                s.indexOf('?')
              else if (s.contains('%'))
                s.indexOf('%')
              else Math.min(100,s.length)
            s.substring(0, endIndex)
          }

          if (pageCode.length >= 4) {
            entranceCode = subField(pageCode.slice(2, 3).head)
            pageTypeCode = subField(pageCode.slice(3, 4).head.trim)
          } else {
            pageTypeCode = subField(pageCode.slice(2, 3).head.trim)
          }

          //map转jsonString
          def map2Json(map : Map[String,Object]) : String = {
            val javamap:java.util.Map[String,Object] = mapAsJavaMap(map).asInstanceOf[java.util.Map[String,Object]]
            val jsonString = JSONObject.toJSONString(javamap)
            jsonString
          }

          //频道页和专题页又前5位固定码和3位随机码生成，在判断页面类型取前5位
          def getPageType(pc: String) = pc match {
            case CHANNEL_REGEX.code(pc) => Map("pageUrl" -> pageUrl, "code" -> pc,"mt" -> parMap.getOrElse("mt", ""),"sk" -> parMap.getOrElse("sk", ""))
            case SEARCH_PRE.code => Map("pageUrl" -> pageUrl, "code" -> SEARCH_PRE.code, "searchKey" -> parMap.getOrElse("sk", ""),
              "keyType" -> parMap.getOrElse("kt", ""))
            case SEARCH_DETAIL.code => {
              var cp = "";
              if (parMap.getOrElse("cp", "").getClass.getSimpleName == "String") {
                cp = parMap.getOrElse("cp", "")
              } else {
                cp = map2Json(parMap.getOrElse("cp", null).asInstanceOf[Map[String, Object]])
              }
              Map("pageUrl" -> pageUrl, "code" -> SEARCH_DETAIL.code, "searchKey" -> parMap.getOrElse("sk", ""),
                "lab" -> parMap.getOrElse("lab", ""), "produceSum" -> parMap.getOrElse("ps", ""),
                "pagination" -> parMap.getOrElse("pag", ""), "filterConditions" -> cp)
            }
            case DETAIL.code => Map("pageUrl" -> pageUrl, "code" -> DETAIL.code, "categoryId" -> parMap.getOrElse("ci", ""), "categoryName" -> parMap.getOrElse("cn", ""),
              "productId" -> parMap.getOrElse("pi", ""), "productName" -> Utils.emojiFilter(parMap.getOrElse("pn", "")), "specialSelling" -> parMap.getOrElse("ss", ""),"strategy" -> parMap.getOrElse("rst","-1"),"board"->parMap.getOrElse("bd","-1"))
            case Fill_ORDER.code => Map("pageUrl" -> pageUrl, "code" -> Fill_ORDER.code, "productId" -> parMap.getOrElse("pi", ""), "categoryId" -> parMap.getOrElse("ci", ""))
            case PENDING_PAYMENT.code => Map("pageUrl" -> pageUrl, "code" -> PENDING_PAYMENT.code,
              "productId" -> parMap.getOrElse("pi", ""), "categoryId" -> parMap.getOrElse("ci", ""),
              "orderNum" -> parMap.getOrElse("on", ""), "paymentAmount" -> parMap.getOrElse("pa", ""))
            case COMPLETED_PAYMENT.code => Map("pageUrl" -> pageUrl, "code" -> COMPLETED_PAYMENT.code, "paymentMethod" -> parMap.getOrElse("pm", ""),
              "successFlag" -> parMap.getOrElse("pson", ""), "paymentAmount" -> parMap.getOrElse("pa", ""), "productId" -> parMap.getOrElse("pi", ""),
              "categoryId" -> parMap.getOrElse("ci", ""), "orderNum" -> parMap.getOrElse("on", ""))
            case TOPIC_REGEX.code(pc)=> {
              var productId = ""
              if(parMap.get("url")!="")
                productId =  regex.findFirstIn(parMap.get("url").toString).getOrElse("")
              Map("pageUrl" -> pageUrl, "code" -> pc,"area" -> parMap.getOrElse("area", ""),
                "navName" -> parMap.getOrElse("navName", ""),"tabName" -> parMap.getOrElse("tabName", ""),"oi" -> parMap.getOrElse("oi", ""),
                "url" -> parMap.getOrElse("url", ""),"productId"->productId)
            }
            case THEME_INDEX.code => Map("board" -> parMap.getOrElse("bd", ""), "categoryId" -> parMap.getOrElse("ci", ""),
              "tagId" -> parMap.getOrElse("cp", ""))
            case THEME_TAG_LIST.code => Map("board" -> parMap.getOrElse("bd", ""), "categoryId" -> parMap.getOrElse("ci", ""),
              "tagId" -> parMap.getOrElse("cp", ""))
            case THEME_GUESS_ENJOY.code => Map("board" -> parMap.getOrElse("bd", ""), "categoryId" -> parMap.getOrElse("ci", ""),
              "tagId" -> parMap.getOrElse("cp", ""))
            case ASSOCIATE_PUSH.code => Map("board" -> parMap.getOrElse("bd", ""), "categoryId" -> parMap.getOrElse("ci", ""),
              "tagId" -> parMap.getOrElse("cp", ""))
            case SUPERFREETRAVEL_HP.code =>{
              var cp="";
              if (parMap.getOrElse("cp", "").getClass.getSimpleName == "String") {
                cp=parMap.getOrElse("cp","")
              } else {
                cp = map2Json(parMap.getOrElse("cp", null).asInstanceOf[Map[String, Object]])
              }
              Map("recommendId" -> parMap.getOrElse("recommendId", ""), "recommendName" -> parMap.getOrElse("recommendName", ""),
                "tagId" -> cp)
            }
            case SUPERFREETRAVEL_TripGuide.code => {
              var cp="";
              if (parMap.getOrElse("cp", "").getClass.getSimpleName == "String") {
                cp=parMap.getOrElse("cp","")
              } else {
                cp = map2Json(parMap.getOrElse("cp", null).asInstanceOf[Map[String, Object]])
              }
              Map("recommendId" -> parMap.getOrElse("recommendId", ""), "recommendName" -> parMap.getOrElse("recommendName", ""),
                "tagId" -> cp)
            }
            case SUPERFREETRAVEL_ChooseResource.code => Map("recommendId" -> parMap.getOrElse("recommendId", ""), "recommendName" -> parMap.getOrElse("recommendName", ""))
            case SUPERFREETRAVEL_Details.code => Map("recommendId" -> parMap.getOrElse("recommendId", ""), "recommendName" -> parMap.getOrElse("recommendName", ""))
            case SUPERFREETRAVEL_ChangeTransp.code => {
              if (parMap.getOrElse("cp", "").getClass.getSimpleName == "String") {
                Map("tagId" -> parMap.getOrElse("cp", ""))
              } else {
                val cp = map2Json(parMap.getOrElse("cp", null).asInstanceOf[Map[String, Object]])
                Map("tagId" -> cp)
              }
            }
            case SUPERFREETRAVEL_HotelList.code =>{
              if(parMap.getOrElse("cp","").getClass.getSimpleName=="String"){
                Map("tagId" -> parMap.getOrElse("cp", ""))
              }else{
                val cp=map2Json(parMap.getOrElse("cp",null).asInstanceOf[Map[String,Object]])
                Map("tagId" -> cp)
              }
            }
//            case SUPER_GO_REGEX.code(pc) => Map("bc" -> parMap.getOrElse("bc", ""), "recommendId" -> parMap.getOrElse("recommendId", ""),
//              "recommendName" -> parMap.getOrElse("recommendName", ""),"cp" -> parMap.getOrElse("cp", ""))

            case _ => Map("pageUrl" -> pageUrl)

          }

          def getPageAttribute(pa: String) = pa match {
            case H5.code => "H5"
            case _ => "NATIVE"
          }

          val (holeCode, accessCode) = entranceCode.split("-").toList match {
            case a :: b :: Nil => (a, b)
            case _ => ("kwnull", "null")
          }

          Some(parMap.getOrElse("lt", ""), parMap.getOrElse("dt", ""), parMap.getOrElse("ot", ""), parMap.getOrElse("nt", ""), parMap.getOrElse("sessionid", ""),
            parMap.getOrElse("lo", ""), parMap.getOrElse("la", ""), parMap.getOrElse("prov", ""), parMap.getOrElse("us", ""),
            parMap.getOrElse("ui", ""), parMap.getOrElse("bc", ""), parMap.getOrElse("pc", ""), channelCode, platformCode, pageTypeCode, getPageType(pageTypeCode),
            parMap.getOrElse("ts", "").toLong, parMap.getOrElse("sti", ""), holeCode, accessCode,
            parMap.getOrElse("nv", ""), ip)
        case None =>
          logger.error(s"WirelessClientMonitorLog None $arg")
          None
        case _ =>
          logger.error(s"WirelessClientMonitorLog an error occurred $arg")
          None
      }
    } catch {
      case e: Exception =>
        logger.error(s"WirelessClientMonitorLog an Exception $arg")
        None
    }
  }
}

/**
  * 实时预处理日志类
  * @param logType
  * @param deviceToken
  * @param sessionId
  * @param province
  * @param userId
  * @param pageCode
  * @param channelCode
  * @param platformCode
  * @param pageTypeCode
  * @param timestamp
  * @param stationId
  */
case class StatLogType(logType: String, deviceToken: String, sessionId: String, province: String, userId: String,
                       pageCode: String, channelCode: String, platformCode: String, pageTypeCode: String,
                       stationId: String, secondChannel: String, categoryId: String, paymentAmount: Double,
                       buttonCode: String, idfa: String, timestamp: Long, useYears: Long, productId: String, searchKey: String)

object StatLogType extends LazyLogging {
  def apply(op: Option[(String, String, String, String, String, String, String, String, String, String, String,
    String, Double, String, String, Long, Long, String, String)]):
  StatLogType = {
    if (op.isDefined) {
      val par = op.get
      new StatLogType(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9, par._10,
        par._11, par._12, par._13, par._14, par._15, par._16, par._17, par._18, par._19)
    } else {
      new StatLogType("", "", "", "", "", "", "", "", "", "", "", "", 0D, "", "", 0L, -1L, "", "")
    }
  }

  def apply(): StatLogType = {
    this.apply("", "", "", "", "", "", "", "", "", "", "", "", 0D, "", "", 0L, -1L, "", "")
  }

  def unapply(arg: String, secondChannel: String, idfa: String): Option[(String, String, String, String, String, String, String, String,
    String, String, String, String, Double, String, String, Long, Long, String, String)] = {
    try {
      JSON.parseFull(arg) match {
        case Some(parMap: HashTrieMap[String, String]) =>
          val pageCode: Array[String] = parMap.get("pc").get.split("\\.")
          val channelCode = pageCode.slice(0, 1).head
          val platformCode = pageCode.slice(1, 2).head
          var entranceCode = "kwnull-null"
          var pageTypeCode = ""
          if (pageCode.length == 4) {
            entranceCode = pageCode.slice(2, 3).head
            pageTypeCode = pageCode.slice(3, 4).head
          } else {
            pageTypeCode = pageCode.slice(2, 3).head
          }

          Some(parMap.getOrElse("lt", ""), parMap.getOrElse("dt", ""), parMap.getOrElse("sessionid", ""),
            parMap.getOrElse("prov", ""), parMap.getOrElse("ui", ""), parMap.getOrElse("pc", ""), channelCode,
            platformCode, pageTypeCode, parMap.getOrElse("sti", ""), secondChannel, parMap.getOrElse("ci", ""),
            parMap.getOrElse("pa", "0").toDouble, parMap.getOrElse("bc", ""), idfa, parMap.getOrElse("ts", "").toLong, -1L,
            parMap.getOrElse("pi", ""), parMap.getOrElse("sk", ""))
        case None =>
          logger.error(s"StatLogType None $arg")
          None
        case _ =>
          logger.error(s"StatLogType an error occurred $arg")
          None
      }
    } catch {
      case e: Exception =>
        logger.error(s"StatLogType an Exception $arg")
        None
    }
  }
}
/**
  * H5页面访问时长监控
  *
  * @param province    省份
  * @param city        城市
  * @param operators   运营商
  * @param networkType 网络类型
  * @param loadTime    页面加载时长
  * @param islv        是否为驴妈妈（区分H5和NATIVE）
  * @param categoryId  品类
  * @param createDate  创建时间
  */
case class H5PageMonitorLog(province: String, city: String, operators: String, networkType: String, loadTime: String,
                            islv: String, categoryId: String, createDate: String, url: String)

object H5PageMonitorLog extends LazyLogging {
  def apply(op: Option[(String, String, String, String, String, String, String, String, String)]): H5PageMonitorLog = {
    if (op.isDefined) {
      val par = op.get
      new H5PageMonitorLog(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9)
    } else {
      new H5PageMonitorLog("", "", "", "", "", "", "", "", "")
    }
  }

  def unapply(arg: String, createDate: String): Option[(String, String, String, String, String, String, String, String, String)] = {
    try {
      val json = JsonParser(arg)
      val jsObj = json.asJsObject()
      val prov = jsObj.getFields("prov").head.toString().replaceAll("\"", "")
      val ct = jsObj.getFields("ct").head.toString().replaceAll("\"", "")
      val ot = jsObj.getFields("ot").head.toString().replaceAll("\"", "")
      val nt = jsObj.getFields("nt").head.toString().replaceAll("\"", "")
      val lt = jsObj.getFields("lt").head.toString().replaceAll("\"", "")
      val islv = jsObj.getFields("islv").head.toString().replaceAll("\"", "")
      val ci = ""
      var url = jsObj.getFields("url").head.toString().replaceAll("\"", "").replace("http://", "").replace("https://", "")
      var index = 0
      if (url.equals("")) {
        None
      }
      index = url.indexOf("?")
      if (index != -1) {
        url = url.substring(0, index)
      }
      if (!lt.matches("\\d*[.|\\d]*")) {
        None
      }
      else if (prov.equals("") || ct.equals("") || ot.equals("") || nt.equals("") || lt.equals("") || islv.equals("") || createDate.length != 19) {
        None
      }
      else {
        Some(prov, ct, ot, nt, lt, islv, ci, createDate, url)
      }
    } catch {
      case e: Exception =>
        logger.error(s"H5PageMonitorLog an error occurred $arg")
        None
    }
  }
}

/**
  *
  * @param operators     运营商
  * @param networkType   网络类型 区分wifi/2g/3g/4g
  * @param deviceToken   设备号
  * @param tec           技术类型(IOS、ANDROID、WAP、H5-IOS、H5-ANDROID)
  * @param version       客户端版本号 如：7.8.1，7.8.2（获取不到传NONE）
  * @param province      省份，获取不到为None
  * @param city      城市，获取不到为None
  * @param mobileNumber      手机号，获取不到为None
  * @param userId      账号，获取不到为None
  * @param networkSpeed  （network speed）公网速度
  * @param osVersion      操作系统
  * @param browser      浏览器
  * @param deviceName  设备型号
  * @param interfaceUniqueFlag     接口唯一标志（时间戳（毫秒）+6位随机数）
  * @param pageId   页面编号（该接口属于哪个页面）
  * @param networkTime  网络耗时（客户端传NONE）
  * @param apiTime       接口内部处理耗时
  * @param apiName       接口名
  * @param apiVersion    接口版本号 版本号为空传NONE
  * @param requestType   请求类型 1:POST 2:GET 3:PATCH 4:PUT 5:DELETE 6:其他
  * @param time           请求时间
  * @param errorCode     接口请求失败对应的异常code放入这边，包括接口异常和HTTP请求异常,如果ec为1则表示请求成功
  */
case class PerformaceMonitorLog(operators: String, networkType: String, deviceToken: String, tec: String, version: String, province: String,city: String,
                                mobileNumber: String, userId: String, networkSpeed: Double,osVersion: String,browser: String,deviceName: String,
                                interfaceUniqueFlag: String, pageId: String,networkTime: Double,apiTime: Double,apiName: String,apiVersion: String,
                                requestType: String, time: Long,errorCode: String)

object PerformaceMonitorLog extends LazyLogging {
  def apply(op: Option[(String, String, String, String, String, String,String, String, String, Double, String, String, String, String,  String, Double, Double, String, String, String, Long,  String)]): PerformaceMonitorLog = {
    if (op.isDefined) {
      val par = op.get
      new PerformaceMonitorLog(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9, par._10, par._11, par._12, par._13, par._14, par._15, par._16, par._17, par._18, par._19, par._20, par._21, par._22)
    } else {
      null
    }
  }

  def process(ot: String, nt: String, dt: String, tec: String, ver: String, prov: String,ct: String, mn: String, ui: String, osVersion: String, deviceName: String,browser: String,firstChannel: String, ns: Double, data: JsonNode, yesterday: Long): PerformaceMonitorLog = {
    try {
      if (dt.equals("")) {
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
      }
      val iuf = data.get("iuf").asText("")
      val pi = data.get("pi").asText("")
      //接口名，版本，请求类型，请求时间部分成功和失败
      var an = data.get("an").asText("")
      val av = data.get("av").asText("")
      val rt = data.get("rt").asText("")
      val method = rt match {
        case "POST" => "1"
        case "GET" => "2"
        case "PATCH " => "3"
        case "PUT" => "4"
        case "DELETE" => "5"
        case _ => "6"
      }
      if (method.equals("6")) {
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
      }
      val time = data.get("time").asLong()
      var ec = data.get("ec").asText("")
      if (ec.equals("1")){
        ec = "200"
      }
      if (an.equals("") || av.equals("") || ec.equals("") || an.equals("NONE")) {
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
      }
      if (an.contains("Api/statusSwitch")) {
        an = "/Api/statusSwitch"
      }
      //昨天之前的数据抛弃
//      if (time < yesterday) {
//        return new PerformaceMonitorLog("", "", "", "", "", "", 0.0, 0.0, 0.0, 0.0, "", 0.0, "", "", "", "", "", 0.0, 0.0, 0.0)
//      }
//      val createTime = Utils.timeFormat(time,"yyyy-MM-dd HH:mm:ss")
      var ft = 0.0
      var tt = 0.0
      var at = 0.0
      //接口总耗时 = 首包时间 + 传输时间 ft + tt
      var totalTime = 0.0
      //接口网络耗时 = 首包时间 + 传输时间 - 接口处理耗时
      var networkTime = 0.0
      //传输速度 = 包大小/传输时间
//      var transferSpeed = 0.0
      //失败标志，1-接口异常，2-HTTP请求异常，0-成功
      var failType = "0"

      /**
        * 接口异常有如下几类
        * 2 - 警告，客户端对该错误码做特殊弹窗处理
        * -1 - 服务器异常
        * -2 - 无产品或者产品不可售
        * -3 - 跳转到登录界面
        * -7 - 无权限
        * -5 - 验证参数失败
        * -101 - 图片验证码失败
        * 3 - 跳转到极验页面
        * 对应的HTTP异常
        * 请提供对应的HTTP请求返回的状态码回来，如404、405
        */
      if ("0".equals(ec) || "NONE".equals(ec)) {
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
      }
      if ("1".equals(ec) || "200".equals(ec)) {
        failType = "0"
      }
      else if ("2".equals(ec) || "-1".equals(ec) || "-2".equals(ec) || "-3".equals(ec) || "-7".equals(ec) || "-5".equals(ec) || "-101".equals(ec) || "3".equals(ec)) {
        failType = "1"
      }
      else {
        failType = "2"
      }

      if ("0".equals(failType) || "1".equals(failType)) {
        at = data.get("at").asDouble()
        if (firstChannel.equals("WAP")){
          networkTime = data.get("nt").asDouble()
        }else{
          ft = data.get("ft").asDouble()
          tt = data.get("tt").asDouble()
          totalTime = ft + tt
          networkTime = totalTime - at
        }
        if (at == 0.0 || at < 0 || networkTime == 0.0 || networkTime < 0 || networkTime > 600000) {
          //            logger.error(an + " :耗时 " + at)
          return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
        }
      }
      new PerformaceMonitorLog(ot, nt, dt, tec, ver, prov,ct,mn,ui, ns,osVersion,browser,deviceName, iuf, pi, networkTime, at,an,av,method,time,ec)
    }
    catch {
      case e: Exception =>
        logger.error("PerformaceMonitorLog an error occurred " + data.toString)
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
    }
  }

  def fullProcess(ot: String, nt: String, dt: String, tec: String, ver: String, prov: String,ct: String, mn: String, ui: String, osVersion: String, deviceName: String,browser: String,firstChannel: String, ns: Double, data: JsonNode, yesterday: Long): PerformaceMonitorLog = {
    try {
      val iuf = data.get("iuf").asText("")
      if (iuf.equals("") || iuf.equals("NONE")) {
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
      }
      val pi = data.get("pi").asText("")
      //接口名，版本，请求类型，请求时间部分成功和失败
      var an = data.get("an").asText("")
      val av = data.get("av").asText("")
      val rt = data.get("rt").asText("")
      val method = rt match {
        case "POST" => "1"
        case "GET" => "2"
        case "PATCH " => "3"
        case "PUT" => "4"
        case "DELETE" => "5"
        case _ => "6"
      }
      val time = data.get("time").asLong()
      var ec = data.get("ec").asText("")
      if (ec.equals("1")){
        ec = "200"
      }
      if (an.contains("Api/statusSwitch")) {
        an = "/Api/statusSwitch"
      }
      //昨天之前的数据抛弃
      //      if (time < yesterday) {
      //        return new PerformaceMonitorLog("", "", "", "", "", "", 0.0, 0.0, 0.0, 0.0, "", 0.0, "", "", "", "", "", 0.0, 0.0, 0.0)
      //      }
      //      val createTime = Utils.timeFormat(time,"yyyy-MM-dd HH:mm:ss")
      var ft = 0.0
      var tt = 0.0
      var at = 0.0
      //接口总耗时 = 首包时间 + 传输时间 ft + tt
      var totalTime = 0.0
      //接口网络耗时 = 首包时间 + 传输时间 - 接口处理耗时
      var networkTime = 0.0
      //传输速度 = 包大小/传输时间
      //      var transferSpeed = 0.0
      //失败标志，1-接口异常，2-HTTP请求异常，0-成功
      var failType = "0"

      /**
        * 接口异常有如下几类
        * 2 - 警告，客户端对该错误码做特殊弹窗处理
        * -1 - 服务器异常
        * -2 - 无产品或者产品不可售
        * -3 - 跳转到登录界面
        * -7 - 无权限
        * -5 - 验证参数失败
        * -101 - 图片验证码失败
        * 3 - 跳转到极验页面
        * 对应的HTTP异常
        * 请提供对应的HTTP请求返回的状态码回来，如404、405
        */
      if ("1".equals(ec) || "200".equals(ec)) {
        failType = "0"
      }
      else if ("2".equals(ec) || "-1".equals(ec) || "-2".equals(ec) || "-3".equals(ec) || "-7".equals(ec) || "-5".equals(ec) || "-101".equals(ec) || "3".equals(ec)) {
        failType = "1"
      }
      else {
        failType = "2"
      }

      if ("0".equals(failType) || "1".equals(failType)) {
        at = data.get("at").asDouble()
        if (firstChannel.equals("WAP")){
          networkTime = data.get("nt").asDouble()
        }else{
          ft = data.get("ft").asDouble()
          tt = data.get("tt").asDouble()
          totalTime = ft + tt
          networkTime = totalTime - at
        }
      }
      new PerformaceMonitorLog(ot, nt, dt, tec, ver, prov,ct,mn,ui, ns,osVersion,browser,deviceName, iuf, pi, networkTime, at,an,av,method,time,ec)
    }
    catch {
      case e: Exception =>
        logger.error("PerformaceMonitorLog an error occurred " + data.toString)
        return new PerformaceMonitorLog("", "", "", "", "", "", "", "", "", 0.0, "", "", "", "", "", 0.0, 0.0, "", "", "", 0l, "")
    }
  }
}

case class MonitorPageLog(url: String, platform: String, createDate: String, whiteScreenTime: Double, firstScreenTime: Double, interactiveTime: Double,
                          downloadTime: Double, dnsTime: Double, buildUnionTime: Double, sslTime: Double, pageLoadTime: Double, backendProcessTime: Double, sslFlag: String)

object MonitorPageLog extends LazyLogging {
  def apply(op: Option[(String, String, String, Double, Double, Double, Double, Double, Double, Double, Double, Double, String)]): MonitorPageLog = {
    if (op.isDefined) {
      val par = op.get
      new MonitorPageLog(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9 , par._10, par._11, par._12, par._13)
    } else {
      new MonitorPageLog("", "", "", 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, "")
    }
  }

  def unapply(arg: String, createTime: String): Option[(String, String, String, Double, Double, Double, Double, Double, Double, Double, Double, Double, String)] = {
    try {
      val json = JsonParser(arg)
      val jsObj = json.asJsObject()
      var url = jsObj.getFields("url").head.toString().replaceAll("\"", "").replace("http://", "").replace("https://", "")
      val pf = jsObj.getFields("pf").head.toString().replaceAll("\"", "")
      val ws = jsObj.getFields("ws").head.toString().replaceAll("\"", "")
      val fs = jsObj.getFields("fs").head.toString().replaceAll("\"", "")
      val it = jsObj.getFields("it").head.toString().replaceAll("\"", "")
      val reqs = jsObj.getFields("reqs").head.toString().replaceAll("\"", "")
      val repe = jsObj.getFields("repe").head.toString().replaceAll("\"", "")
      val reps = jsObj.getFields("reps").head.toString().replaceAll("\"", "")
      val dle = jsObj.getFields("dle").head.toString().replaceAll("\"", "")
      val dls = jsObj.getFields("dls").head.toString().replaceAll("\"", "")
      val cs = jsObj.getFields("cs").head.toString().replaceAll("\"", "")
      val ce = jsObj.getFields("ce").head.toString().replaceAll("\"", "")
      val ss = jsObj.getFields("ss").head.toString().replaceAll("\"", "")
      val le = jsObj.getFields("le").head.toString().replaceAll("\"", "")
      val ns = jsObj.getFields("ns").head.toString().replaceAll("\"", "")

      var index = 0
      if (url.equals("")) {
        None
      }
      index = url.indexOf("?")
      if (index != -1) {
        url = url.substring(0, index)
      }
      if(!url.contains("lvmama") && !url.contains("localhost") && !url.contains("file") && !url.contains("*.*.*.*:*")){
        None
      }
      if (pf.equals("") ||ws.equals("") ||fs.equals("") ||it.equals("") ||reqs.equals("") || repe.equals("") || reps.equals("") || dle.equals("") || dls.equals("") || cs.equals("") || ce.equals("") || ss.equals("") || le.equals("") || ns.equals("") || createTime.length != 19 || createTime.equals("")) {
        None
      }
      else {
        //白屏时间
        val whiteScreenTime = ws.toDouble./(1000)
        //首屏时间
        val firstScreenTime = fs.toDouble./(1000)
        //可交互时间
        val interactiveTime = it.toDouble./(1000)
        //下载时间
        val downloadTime = (repe.toDouble - reps.toDouble)./(1000)
        //DNS时间
        val dnsTime = (dle.toDouble - dls.toDouble)./(1000)
        //建联时间
        val buildUnionTime = (ce.toDouble - cs.toDouble)./(1000)
        //SSL时间
        var sslTime = 0.0
        var sslFlag = "0"
        if (!ss.equals("NONE")){
          sslTime = (ce.toDouble - ss.toDouble)./(1000)
          sslFlag = "1"
        }
        //页面加载完成时间
        val pageLoadTime = (le.toDouble - ns.toDouble)./(1000)
        //后端时间
        val backendProcessTime = (reps.toDouble - reqs.toDouble)./(1000)
        //操作时间
        val createDate = createTime.substring(0,10)
        if (whiteScreenTime < 0.0 || firstScreenTime < 0.0 || interactiveTime < 0.0 || downloadTime <0.0 || dnsTime <0.0 || buildUnionTime < 0.0
          || sslTime <0.0 || pageLoadTime<0.0 || backendProcessTime < 0.0){
          None
        }else {
          Some(url, pf, createDate, whiteScreenTime, firstScreenTime, interactiveTime, downloadTime, dnsTime, buildUnionTime, sslTime, pageLoadTime, backendProcessTime, sslFlag)
        }
      }
    } catch {
      case e: Exception =>
        logger.error(s"MonitorPageLog an error occurred $arg")
        None
    }
  }
}


case class MonitorPageInfoLog(url: String, pid: String, tec: String, time: Long, userInformation: Map[String,String], deviceInformation: Map[String,String], userTime: Map[String,Long],
                              events: Map[String,Long], threshold:Map[String,String])

object MonitorPageInfoLog extends LazyLogging {
  def apply(op: Option[(String, String, String, Long, Map[String,String], Map[String,String], Map[String,Long], Map[String,Long], Map[String,String])]): MonitorPageInfoLog = {
    if (op.isDefined) {
      val par = op.get
      new MonitorPageInfoLog(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9)
    } else {
      new MonitorPageInfoLog("", "", "", 0l, null, null, null, null, null)
    }
  }

  def process(arg: String, createTime: String, threshold: Map[Any,Row]): MonitorPageInfoLog = {
    try {
      val mapper = new ObjectMapper()
      val jsonNode = mapper.readTree(arg)
      //url
      val url = jsonNode.path("url").asText("")
      //pid
      val pid = jsonNode.path("pid").asText("")
      //技术类型
      val tec = jsonNode.path("tec").asText("")

      //请求时间戳
      val time = Utils.date2Stamp(createTime)

      //用户信息
      val ui = jsonNode.findPath("ui")
      //ip
      val ip = ui.path("ip").asText("")
      //运营商
      val operators = ui.path("ot").asText("")
      //网络类型 区分wifi/2g/3g/4g
      val networkType = ui.path("nt").asText("")
      //城市
      val city = ui.path("ct").asText("")
      //省份
      val province = ui.path("prov").asText("")
      //经度
      val latitude = ui.path("la").asText("")
      //维度
      val longtitude = ui.path("lo").asText("")
      //手机号
      val mobileNumber = ui.path("mn").asText("")
      //用户账号
      val usersId = ui.path("ui").asText("")

      //用户信息map
      var userInformation = Map[String, String]()
      userInformation += ("ip" -> ip)
      userInformation += ("operators" -> operators)
      userInformation += ("networkType" -> networkType)
      userInformation += ("city" -> city)
      userInformation += ("province" -> province)
      userInformation += ("latitude" -> latitude)
      userInformation += ("longtitude" -> longtitude)
      userInformation += ("mobileNumber" -> mobileNumber)
      userInformation += ("usersId" -> usersId)


      //设备信息
      val di = jsonNode.findPath("di")
      //分辨率
      val resolution = di.path("resl").asText("")
      //设备型号
      val deviceName = di.path("dm").asText("")
      //设备号
      val deviceToken = di.path("dt").asText("")
      //操作系统
      val osVersion = di.path("os").asText("")
      //ua
      val ua = di.path("ua").asText("")

      //用户信息map
      var deviceInformation = Map[String, String]()
      deviceInformation += ("resolution" -> resolution)
      deviceInformation += ("deviceName" -> deviceName)
      deviceInformation += ("deviceToken" -> deviceToken)
      deviceInformation += ("osVersion" -> osVersion)
      deviceInformation += ("ua" -> ua)

      //用户时间
      val ut = jsonNode.findPath("ut")
      //白屏时间
      val whiteScreenTime = ut.path("ws").asLong()
      //首屏时间
      val firstScreenTime = ut.path("fs").asLong()
      //可交互时间
      val interactiveTime = ut.path("it").asLong()
      //完成加载时间
      val completeLoadTime = ut.path("comp").asLong()

      //是否为慢白屏（true/false）
      var isWS = "false"
      //是否为慢首屏（true/false）
      var isFS = "false"
      //是否为慢加载（true/false）
      var isCL = "false"
      if (tec.equals("WAP") || tec.equals("H5-IOS") || tec.equals("H5-ANDROID")) {
        val row = threshold.get("WAP").get
        val abnormalData = row(5).asInstanceOf[Integer].toLong
        if (whiteScreenTime > abnormalData || firstScreenTime > abnormalData || completeLoadTime > abnormalData) {
          return new MonitorPageInfoLog("", "", "", 0l, null, null, null, null, null)
        }
        if (whiteScreenTime > row(3).asInstanceOf[Integer].toLong) {
          isWS = "true"
        }
        if (firstScreenTime > row(2).asInstanceOf[Integer].toLong) {
          isFS = "true"
        }
        if (completeLoadTime > row(1).asInstanceOf[Integer].toLong) {
          isCL = "true"
        }
      }
      if (tec.equals("PC")) {
        val row = threshold.get("PC").get
        val abnormalData = row(5).asInstanceOf[Integer].toLong
        if (whiteScreenTime > abnormalData || firstScreenTime > abnormalData || completeLoadTime > abnormalData) {
          return new MonitorPageInfoLog("", "", "", 0l, null, null, null, null, null)
        }
        if (whiteScreenTime > row(3).asInstanceOf[Integer].toLong) {
          isWS = "true"
        }
        if (firstScreenTime > row(2).asInstanceOf[Integer].toLong) {
          isFS = "true"
        }
        if (completeLoadTime > row(1).asInstanceOf[Integer].toLong) {
          isCL = "true"
        }
      }

      //慢阀值map
      var thresholdMap = Map[String, String]()
      thresholdMap += ("isWS" -> isWS)
      thresholdMap += ("isFS" -> isFS)
      thresholdMap += ("isCL" -> isCL)

      //用户时间map
      var userTime = Map[String, Long]()
      userTime += ("whiteScreenTime" -> whiteScreenTime)
      userTime += ("firstScreenTime" -> firstScreenTime)
      userTime += ("interactiveTime" -> interactiveTime)
      userTime += ("completeLoadTime" -> completeLoadTime)

      //请求流
      val es = jsonNode.findPath("es")
      //为浏览器发送从服务器或者缓存获取实际文档的请求之时的 Unix毫秒时间戳
      val requestStart = es.path("reqs").asLong()
      //浏览器从服务器、缓存或者本地资源接收响应的最后一个字节或者连接被关闭之时的 Unix毫秒时间戳
      val responseEnd = es.path("repe").asLong()
      //为浏览器从服务器、缓存或者本地资源接收到响应的第一个字节之时的 Unix毫秒时间戳
      val responseStart = es.path("reps").asLong()
      //解析域名结束时的 Unix毫秒时间戳
      val domainLookupEnd = es.path("dle").asLong()
      //为域名开始解析之时的 Unix毫秒时间戳
      val domainLookupStart = es.path("dls").asLong()
      //请求连接被发送到网络之时的Unix毫秒时间戳
      val connectStart = es.path("cs").asLong()
      //网络链接建立的时间节点
      val connectEnd = es.path("ce").asLong()
      //安全连接握手开始的时刻的 Unix毫秒时间戳
      val secureConnectionStart = es.path("ss").asLong()
      //为 load 事件处理程序被终止，加载事件已经完成之时的 Unix毫秒时间戳
      val loadEventEnd = es.path("le").asLong()
      //紧接着在相同的浏览环境下卸载前一个文档结束之时的 Unix毫秒时间戳
      val navigationStart = es.path("ns").asLong()
      //unload
      val unload = es.path("ul").asLong()
      //重定向开始时间
      val redirectStart = es.path("rs").asLong()
      //重定向结束时间
      val redirectEnd = es.path("re").asLong()
      //dom加载时间
      val domLoading = es.path("dl").asLong()
      //加载事件开始时间
      val loadEventStart = es.path("les").asLong()

      //请求流map
      var events = Map[String, Long]()
      events += ("requestStart" -> requestStart)
      events += ("responseEnd" -> responseEnd)
      events += ("responseStart" -> responseStart)
      events += ("domainLookupEnd" -> domainLookupEnd)
      events += ("domainLookupStart" -> domainLookupStart)
      events += ("connectStart" -> connectStart)
      events += ("connectEnd" -> connectEnd)
      events += ("secureConnectionStart" -> secureConnectionStart)
      events += ("loadEventEnd" -> loadEventEnd)
      events += ("navigationStart" -> navigationStart)
      events += ("unload" -> unload)
      events += ("redirectStart" -> redirectStart)
      events += ("redirectEnd" -> redirectEnd)
      events += ("domLoading" -> domLoading)
      events += ("loadEventStart" -> loadEventStart)

      new MonitorPageInfoLog(url, pid, tec, time, userInformation, deviceInformation, userTime, events, thresholdMap)
    }catch {
      case e: Exception =>
        logger.error("MonitorPageInfoLog an error occurred " + arg.toString)
        return new MonitorPageInfoLog("", "", "", 0l, null, null, null, null, null)
    }
  }
}


case class RequestFlow(url: String, pid: String, time:Long, name: String, resourceType: String, duration: Long, decodeBodySize: Long, startTime: Long,
                       responseTime: Long, isSlow: String)

object RequestFlow extends LazyLogging {
  def apply(op: Option[(String, String, Long, String, String, Long, Long, Long, Long, String)]): RequestFlow = {
    if (op.isDefined) {
      val par = op.get
      new RequestFlow(par._1, par._2, par._3, par._4, par._5, par._6, par._7, par._8, par._9, par._10)
    } else {
      new RequestFlow("", "", 0l, "", "", 0l, 0l, 0l, 0l, "")
    }
  }

  def preRequestFlow(arg: String, createTime: String): (String, String, String, Long, Array[JsonNode]) = {
    val emptyArray = new Array[JsonNode](0)
    try {
      val mapper = new ObjectMapper()
      val jsonNode = mapper.readTree(arg)
      //url
      val url = jsonNode.path("url").asText("")
      //pid
      val pid = jsonNode.path("pid").asText("")
      //tec
      val tec = jsonNode.path("tec").asText("")

      //请求时间戳
      val time = Utils.date2Stamp(createTime)

      val emptyArray = new Array[JsonNode](0)
      //接口请求信息jsonarray
      val data = jsonNode.path("rs")
      if (data.size() == 0) {
        return ("", "", "", 0l, emptyArray)
      }
      val array = new Array[JsonNode](data.size())
      var i = 0;
      while (i < data.size()) {
        array(i) = data.get(i)
        i = i + 1
      }
      (url, pid, tec, time, array)
    }catch {
      case e: Exception =>
        logger.error(s"preRequestFlow an error")
        ("", "", "", 0l, emptyArray)
    }
  }

  def process(url: String, pid: String, tec:String, time:Long, data: JsonNode, threshold: Map[Any,Row]): RequestFlow = {
    try {
      val name = data.get("na").asText("")
      val resourceType = data.get("tp").asText("")
      val duration = data.get("dur").asLong()
      val decodeBodySize = data.get("dbs").asLong()
      val startTime = data.get("st").asLong()
      val responseTime = data.get("rt").asLong()

      var isSlow = "false"
      if (tec.equals("WAP") || tec.equals("H5-IOS") || tec.equals("H5-ANDROID")) {
        val row = threshold.get("WAP").get
        if (duration > row(4).asInstanceOf[Integer].toLong) {
          isSlow = "true"
        }
      }
      if (tec.equals("PC")) {
        val row = threshold.get("PC").get
        if (duration > row(4).asInstanceOf[Integer].toLong) {
          isSlow = "true"
        }
      }
      new RequestFlow(url, pid, time, name, resourceType, duration, decodeBodySize, startTime, responseTime, isSlow)
    }catch {
      case e: Exception =>
        logger.error("RequestFlow an error occurred " + data.toString)
        return new RequestFlow("", "", 0l, "", "", 0l, 0l, 0l, 0l, "")
    }
  }
}


case class JsError(url: String, pid: String, jsName: String, errorType: String, message: String, stack: String)

object JsError extends LazyLogging {
  def apply(op: Option[(String, String, String, String, String, String)]): JsError = {
    if (op.isDefined) {
      val par = op.get
      new JsError(par._1, par._2, par._3, par._4, par._5, par._6)
    } else {
      new JsError("", "", "", "", "", "")
    }
  }

  def preJsError(arg: String): (String, String, Array[JsonNode]) = {
    val emptyArray = new Array[JsonNode](0)
    try{
      val mapper = new ObjectMapper()
      val jsonNode = mapper.readTree(arg)
      //url
      val url = jsonNode.path("url").asText("")
      //pid
      val pid = jsonNode.path("pid").asText("")

      val emptyArray = new Array[JsonNode](0)
      //接口请求信息jsonarray
      val data = jsonNode.path("data")
      if (data.size() == 0) {
        return ("", "", emptyArray)
      }
      val array = new Array[JsonNode](data.size())
      var i = 0;
      while (i < data.size() ) {
        array(i) = data.get(i)
        i = i + 1
      }
      (url, pid, array)
    }catch {
      case e: Exception =>
        logger.error(s"preJsError an error")
        ("", "", emptyArray)
    }
  }

  def process(url: String, pid: String, data: JsonNode): JsError = {
    try {
      val jsName = data.get("jn").asText("")
      val errorType = data.get("et").asText("")
      val message = data.get("msg").asText("")
      val stack = data.get("stk").asText("")
      val wholeStack = errorType + " : " + message + "  " + stack

      new JsError(url, pid, jsName, errorType, message, wholeStack)
    }catch {
      case e: Exception =>
        logger.error("JsError an error occurred " + data.toString)
        return new JsError("", "", "", "", "", "")
    }
  }
}